package org.jetlinks.pro.rule.engine.cluster;

import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.jetlinks.pro.rule.engine.cluster.strategies.AllSchedulerSelectorStrategy;
import org.jetlinks.pro.rule.engine.cluster.strategies.MinimumLoadSchedulerSelectorStrategy;
import org.jetlinks.rule.engine.api.scheduler.ScheduleJob;
import org.jetlinks.rule.engine.api.scheduler.Scheduler;
import org.jetlinks.rule.engine.api.scheduler.SchedulerSelector;
import org.jetlinks.rule.engine.cluster.SchedulerRegistry;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import javax.annotation.Nonnull;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 集群调度器选择器,根据策略选择对应的调度器来执行对应的规则。可通过实现接口{@link SchedulerSelectorStrategy}并注入到Spring
 * 来实现自定义的选择策略
 *
 * @author zhouhao
 * @since 1.0
 */
@Slf4j
@AllArgsConstructor
public class ClusterSchedulerSelector implements SchedulerSelector {

    private static final SchedulerSelectorStrategy defaultStrategy = new AllSchedulerSelectorStrategy();

    private final Map<String, SchedulerSelectorStrategy> strategies = new ConcurrentHashMap<>();

    public void addStrategy(SchedulerSelectorStrategy strategy) {
        strategies.put(strategy.getType(), strategy);
    }

    @Override
    public Flux<Scheduler> select(Flux<Scheduler> schedulers, ScheduleJob job) {
        //没指定调度规则,使用默认
        SchedulerSelectorStrategy strategy = getStrategy(job);
        return strategy
            .select(schedulers, job)
            .doOnNext(scheduler -> log
                .info("select scheduler[{}] for [{}-{}] using {}",
                      scheduler.getId(),
                      job.getInstanceId(),
                      job.getNodeId(),
                      strategy.getType()));
    }

    @Override
    public Mono<Boolean> test(Scheduler scheduler, ScheduleJob job) {
        return getStrategy(job).test(scheduler, job);
    }

    private SchedulerSelectorStrategy getStrategy(ScheduleJob job) {
        //没指定调度规则,使用默认
        if (job.getSchedulingRule() == null) {
            return defaultStrategy;
        }
        return Optional
            .of(strategies.get(job.getSchedulingRule().getType()))
            .orElse(defaultStrategy);
    }

}
